package io.kiki.sba.registry.server.data_store.resource;


import io.kiki.sba.registry.common.model.ClientInterfaceId;
import io.kiki.sba.registry.common.model.Response;
import io.kiki.sba.registry.common.model.ServerData;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.dataserver.BatchRequest;
import io.kiki.sba.registry.common.model.dataserver.Datum;
import io.kiki.sba.registry.common.model.slot.Slot;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.common.model.store.UnPublisher;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.server.data_store.slot.SlotManager;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.server.shared.remoting.AbstractServerHandler;
import io.kiki.sba.registry.server.shared.util.DatumUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.core.MediaType;
import java.util.*;


@Path("datum/api")
public class DatumApiResource {

    private static final Logger logger = LoggerFactory.getLogger(DatumApiResource.class);

    private static final ClientInterfaceId TEMP_SESSION_PROCESSID = ServerEnv.PROCESS_ID;

    @Autowired
    ServerConfig serverConfig;

    @Autowired
    DatumStorageDelegate datumStorageDelegate;

    @Autowired
    AbstractServerHandler batchPutDataHandler;

    @Autowired
    SlotManager slotManager;

    /**
     * curl -i -d '{"dataInfoId":"testDataId#@#DEFAULT_INSTANCE_ID#@#DEFAULT_GROUP"}' -H
     * "Content-Type: application/json" -X POST http://localhost:9622/datum/api/get
     *
     * @param datumParam datumParam
     * @return CommonResponse
     */
    @POST
    @Path("/get")
    @Produces(MediaType.APPLICATION_JSON)
    public Response get(DatumParam datumParam) {
        try {
            validateAndCorrect(datumParam);
        } catch (RuntimeException e) {
            logger.error(e.getMessage(), e);
            return Response.buildFailedResponse(e.getMessage());
        }

        Datum datum = datumStorageDelegate.get(datumParam.getDataCenter(), datumParam.getDataInfoId());
        if (datum == null) {
            return getNotFoundResponse(datumParam);
        }

        return createResponse(datum);
    }

    @GET
    @Path("/summary/query")
    @Produces(MediaType.APPLICATION_JSON)
    public Map<String, Integer> summary(String dataCenter) {
        if (StringUtils.isBlank(dataCenter)) {
            dataCenter = serverConfig.getDataCenter();
        }

        Map<String, Map<String, Integer>> pubCount = datumStorageDelegate.getLocalPubCount();
        return pubCount.get(dataCenter);
    }

    /**
     * curl -i -d '{"dataInfoId":"testDataId#@#DEFAULT_INSTANCE_ID#@#DEFAULT_GROUP"}' -H
     * "Content-Type: application/json" -X POST http://localhost:9622/datum/api/delete
     *
     * @param datumParam datumParam
     * @return CommonResponse
     */
    @POST
    @Path("/delete")
    @Produces(MediaType.APPLICATION_JSON)
    public Response delete(DatumParam datumParam) {
        try {
            checkTestApiEnable();
            validateAndCorrect(datumParam);
        } catch (RuntimeException e) {
            logger.error(e.getMessage(), e);
            return Response.buildFailedResponse(e.getMessage());
        }

        Datum datum = datumStorageDelegate.get(datumParam.getDataCenter(), datumParam.getDataInfoId());
        if (datum == null) {
            return getNotFoundResponse(datumParam);
        }
        datumStorageDelegate.cleanLocal(datumParam.getDataCenter(), datumParam.getDataInfoId());
        // get the newly datum
        datum = datumStorageDelegate.get(datumParam.getDataCenter(), datumParam.getDataInfoId());
        return createResponse(datum);
    }

    /**
     * curl -i -d '{"dataInfoId":"testDataId#@#DEFAULT_INSTANCE_ID#@#DEFAULT_GROUP",
     * "publisherDataBox":"1234", "publisherCell":"TestZone2", "publisherConnectId":"127.0.0.1:80"} '
     * -H "Content-Type: application/json" -X POST http://localhost:9622/datum/api/pub/add curl -i -d
     * '{"dataInfoId":"testDataId#@#DEFAULT_INSTANCE_ID#@#DEFAULT_GROUP", "publisherDataBox":"1234",
     * "publisherCell":"TestZone2"} ' -H "Content-Type: application/json" -X POST
     * http://localhost:9622/datum/api/pub/add "publisherConnectId" can be unspecified. If not
     * specified, the connectId of the first pub under the datum can be automatically selected (for
     * testing purposes, it is generally not necessary to specify the connectId artificially).
     *
     * @param datumParam datumParam
     * @return CommonResponse
     */
    @POST
    @Path("/pub/add")
    @Produces(MediaType.APPLICATION_JSON)
    public Response addPub(DatumParam datumParam) {
        try {
            checkTestApiEnable();
            validateAndCorrect(datumParam);

            // build pub
            Datum datum = datumStorageDelegate.get(datumParam.getDataCenter(), datumParam.getDataInfoId());
            if (datum == null) {
                return getNotFoundResponse(datumParam);
            }
            Publisher publisher = buildPublisher(datum, datumParam);

            // build request and invoke
            final int slotId = slotManager.slotOf(publisher.getDataInfoId());
            final Slot slot = slotManager.getSlot(serverConfig.getDataCenter(), slotId);
            BatchRequest batchRequest = new BatchRequest(publisher.getClientInterfaceId(), slotId, Collections.singletonList(publisher));
            batchRequest.setSlotsEpoch(slotManager.getSlotsEpoch());
            batchRequest.setSlotLeaderEpoch(slot.getLeaderEpoch());
            batchPutDataHandler.doHandle(null, batchRequest);
            // get the newly datum
            datum = datumStorageDelegate.get(datumParam.getDataCenter(), datumParam.getDataInfoId());
            return createResponse(datum);
        } catch (RuntimeException e) {
            logger.error(e.getMessage(), e);
            return Response.buildFailedResponse(e.getMessage());
        }
    }

    /**
     * curl -i -d '{"dataInfoId":"testDataId#@#DEFAULT_INSTANCE_ID#@#DEFAULT_GROUP",
     * "publisherRegisterId":"98de0e41-2a4d-44d7-b1f7-c520660657e8"} ' -H "Content-Type:
     * application/json" -X POST http://localhost:9622/datum/api/pub/delete
     *
     * @param datumParam datumParam
     * @return CommonResponse
     */
    @POST
    @Path("/pub/delete")
    @Produces(MediaType.APPLICATION_JSON)
    public Response deletePub(DatumParam datumParam) {
        try {
            checkTestApiEnable();
            validateAndCorrect(datumParam);

            // build pub
            Datum datum = datumStorageDelegate.get(datumParam.getDataCenter(), datumParam.getDataInfoId());
            if (datum == null) {
                return getNotFoundResponse(datumParam);
            }
            UnPublisher publisher = buildUnPublisher(datum, datumParam);

            // build request and invoke
            final int slotId = slotManager.slotOf(publisher.getDataInfoId());
            final Slot slot = slotManager.getSlot(serverConfig.getDataCenter(), slotId);
            BatchRequest batchRequest = new BatchRequest(publisher.getClientInterfaceId(), slotId, Collections.singletonList(publisher));
            batchRequest.setSlotsEpoch(slotManager.getSlotsEpoch());
            batchRequest.setSlotLeaderEpoch(slot.getLeaderEpoch());
            batchPutDataHandler.doHandle(null, batchRequest);
            // get the newly datum
            datum = datumStorageDelegate.get(datumParam.getDataCenter(), datumParam.getDataInfoId());
            return createResponse(datum);
        } catch (RuntimeException e) {
            logger.error(e.getMessage(), e);
            return Response.buildFailedResponse(e.getMessage());
        }
    }

    /**
     * curl -d '{"dataCenter":"registry-cloud-test-b"}' -H "Content-Type: application/json"
     * http://localhost:9622/datum/api/getDatumVersions
     *
     * @param datumParam datumParam
     * @return Object
     */
    @POST
    @Path("/getDatumVersions")
    @Produces(MediaType.APPLICATION_JSON)
    public Object getDatumVersions(DatumParam datumParam) {
        Map<String, Long> datumVersions = _getDatumVersions(datumParam);

        return datumVersions;
    }

    private Map<String, Long> _getDatumVersions(DatumParam datumParam) {
        Map<String, Long> datumVersions = new HashMap<>();
        if (serverConfig.isLocalDataCenter(datumParam.getDataCenter())) {
            Map<String, Datum> dataInfoIdToDatumMap = datumStorageDelegate.getDataInfoIdToDatumMap(serverConfig.getDataCenter());
            return DatumUtils.getDataInfoIdToDatumVersionMap(dataInfoIdToDatumMap);
        } else {
            // TODO need support remote datecenter
            logger.error("unsupport remote datacenter, {}", datumParam.getDataCenter());
        }
        return datumVersions;
    }

    /**
     * curl -d '{"dataCenter":"registry-cloud-test-b"}' -H "Content-Type: application/json"
     * http://localhost:9622/datum/api/getDatumVersions
     *
     * @param datumParam datumParam
     * @return Object
     */
    @POST
    @Path("/getRemoteDatumVersions")
    @Produces(MediaType.APPLICATION_JSON)
    public Object getRemoteDatumVersions(DatumParam datumParam) {
        throw new UnsupportedOperationException();
    }

    /**
     * curl -H "Content-Type: application/json" http://localhost:9622/datum/api/getDatumSizes
     *
     * @return Object
     */
    @GET
    @Path("/getDatumSizes")
    @Produces(MediaType.APPLICATION_JSON)
    public Object getDatumSizes() {
        Map<String, Integer> datumSizes = new HashMap<>();

        Map<String, Datum> localDatums = datumStorageDelegate.getDataInfoIdToDatumMap(serverConfig.getDataCenter());
        int localDatumSize = localDatums.size();
        datumSizes.put(serverConfig.getDataCenter(), localDatumSize);
        // TODO remote cluster
        return datumSizes;
    }

    /**
     * curl -d '{"dataCenter":"registry-cloud-test-b",
     * "dataInfoId":"testDataId#@#DEFAULT_INSTANCE_ID#@#DEFAULT_GROUP"}' -H "Content-Type:
     * application/json" http://localhost:9622/datum/api/getDatumVersion
     *
     * @param datumParam datumParam
     * @return Object
     */
    @POST
    @Path("/getDatumVersion")
    @Produces(MediaType.APPLICATION_JSON)
    public Object getDatumVersion(DatumParam datumParam) {
        String format = "dataCenter:%s, dataInfoId:%s, dataServer:%s, version:%s";
        String dataServer = null;
        Long version = null;
        if (serverConfig.isLocalDataCenter(datumParam.getDataCenter())) {
            Map<String, Datum> localDatums = datumStorageDelegate.getDataInfoIdToDatumMap(serverConfig.getDataCenter());
            Datum datum = localDatums.get(datumParam.getDataInfoId());
            if (datum != null) {
                version = datum.getVersion();
            }
        } else {
            // TODO unsupport remote cluster
            version = null;
        }

        return String.format(format, datumParam.getDataCenter(), datumParam.getDataInfoId(), dataServer, version);
    }

    private Response getNotFoundResponse(DatumParam datumParam) {
        return Response.buildFailedResponse(String.format("datum not found by dataCenter=%s, dataInfoId=%s", datumParam.getDataCenter(), datumParam.getDataInfoId()));
    }

    private void validateAndCorrect(DatumParam datumParam) {
        Validate.notNull(datumParam, "datumParam is null");
        Validate.notEmpty(datumParam.getDataInfoId(), "datumParam.dataInfoId is empty");
        if (StringUtils.isBlank(datumParam.getDataCenter())) {
            datumParam.setDataCenter(serverConfig.getDataCenter());
        }
    }

    Publisher buildPublisher(Datum datum, DatumParam datumParam) {
        Validate.notNull(datumParam.getPublisherRegisterId(), "datumParam.publisherRegisterId is null");
        Validate.notNull(datumParam.getPublisherCell(), "datumParam.publisherCell is null");
        Validate.notNull(datumParam.getPublisherDataBox(), "datumParam.publisherDataBox is null");

        Publisher publisher = new Publisher();
        publisher.setClientInterfaceId(TEMP_SESSION_PROCESSID);
        publisher.setDataInfoId(datumParam.getDataInfoId());
        publisher.setRegisterId(datumParam.getPublisherRegisterId());
        publisher.setRegisterTimestamp(datumParam.getPublisherRegisterTimestamp());
        publisher.setVersion(datumParam.getPublisherVersion());
        if (datumParam.getPublisherConnectId() != null) {
            String[] parts = datumParam.getPublisherConnectId().split(ValueConstants.CONNECT_ID_SPLIT);
            if (parts.length < 2) {
                throw new IllegalArgumentException("datumParam.publisherConnectId format invalid, should be clientIp:clientPort_sessionIp:sessionPort");
            }
            publisher.setClientUrl(Url.valueOf(parts[0]));
            publisher.setServerUrl(Url.valueOf(parts[1]));
        } else {
            Collection<Publisher> publishers = datum.getRegisterIdToPublisherMap().values();
            if (publishers.size() == 0) {
                throw new IllegalArgumentException("datumParam.publisherConnectId is null, and datum has no pubs");
            }
            Publisher firstPub = publishers.iterator().next();
            publisher.setClientUrl(firstPub.getClientUrl());
            publisher.setServerUrl(firstPub.getServerUrl());
        }
        publisher.setZone(datumParam.getPublisherCell());
        ServerData dataBox = new ServerData(datumParam.getPublisherDataBox());
        List<ServerData> dataList = new ArrayList<>(Arrays.asList(dataBox));
        publisher.setServerDataList(dataList);
        return publisher;
    }

    private UnPublisher buildUnPublisher(Datum datum, DatumParam datumParam) {
        Validate.notNull(datumParam.getPublisherRegisterId(), "datumParam.publisherRegisterId is null");

        UnPublisher publisher = new UnPublisher(datumParam.getDataInfoId(), TEMP_SESSION_PROCESSID, datumParam.getPublisherRegisterId(), datumParam.getPublisherRegisterTimestamp(), datumParam.getPublisherVersion());

        Publisher _publisher = datum.getRegisterIdToPublisherMap().get(datumParam.getPublisherRegisterId());
        if (_publisher == null) {
            throw new IllegalArgumentException(String.format("No pub of registerId(%s)", datumParam.getPublisherRegisterId()));
        }
        publisher.setClientUrl(_publisher.getClientUrl());
        publisher.setServerUrl(_publisher.getServerUrl());

        return publisher;
    }

    private void checkTestApiEnable() {
        Validate.isTrue(serverConfig.isEnableTestApi(), "This api is disabled");
    }

    private Response createResponse(Datum datum) {
        return Response.buildSuccessResponse(String.format("datum=%s, publishers=%s", datum, datum.getRegisterIdToPublisherMap()));
    }
}
